package com.west.xu.tdmq.handle;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.MessageListener;

public class ConsumerDeadLetterMsg<T> implements MessageListener<T> {

    private static final Log logger = LogFactory.getLog(ConsumerDeadLetterMsg.class);

    @Override
    public void received(Consumer<T> consumer, Message<T> message) {
        try {
//            String jsonMessageId = new GsonBuilder().create().toJson(messageId.getMessageId());
            String messageId = MessageId.fromByteArray(message.getMessageId().toByteArray()).toString();
            String msg = new String(message.getData());
            logger.info("接收到死信消息===>" + msg + "|MessageId===>" + messageId);
            consumer.acknowledge(message);
            logger.info("接收死信消息处理成功，应答成功");
        } catch (Exception e) {
            consumer.negativeAcknowledge(message);
            logger.error("接收死信消息异常，拒绝应答|Ex==" + e.getMessage());
        }
    }
}
